package com.xrui.hbase;

import com.google.common.base.Preconditions;
import com.xrui.hbase.codec.CellDecoder;
import com.xrui.hbase.codec.CellDecoderFactory;
import com.xrui.hbase.codec.impl.ProtobufCellDecoder;
import com.xrui.hbase.codec.impl.RawBytesCellDecoder;
import com.xrui.hbase.codec.impl.SpecificAvroCellDecoder;
import com.xrui.hbase.impl.BoundColumnReaderSpec;
import com.xrui.hbase.schema.TableLayout;
import com.xrui.hbase.schema.spec.CellSpec;


import java.io.IOException;

/**
 * Factory for HBase cell decoders using SpecificCellDecoder to handle record-based schemas.
 */
public final class SpecificCellDecoderFactory implements CellDecoderFactory {
    /**
     * Singleton instance.
     */
    private static final SpecificCellDecoderFactory SINGLETON = new SpecificCellDecoderFactory();

    /**
     * Singleton constructor.
     */
    private SpecificCellDecoderFactory() {
    }

    /**
     * @return an instance of the SpecificCellDecoderFactory.
     */
    public static CellDecoderFactory get() {
        return SINGLETON;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> CellDecoder<T> create(CellSpec cellSpec) throws IOException {
        Preconditions.checkNotNull(cellSpec);
        switch (cellSpec.getCellSchema().getType()) {
            case AVRO:
            case PROTOBUF:
                return new ProtobufCellDecoder<T>(cellSpec);
            case RAW_BYTES:
                return new RawBytesCellDecoder(cellSpec);
            default:
                throw new InternalError("Unhandled cell encoding: " + cellSpec.getCellSchema());
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> CellDecoder<T> create(TableLayout layout, BoundColumnReaderSpec spec)
        throws IOException {
        Preconditions.checkNotNull(spec);
        switch (spec.getColumnReaderSpec().getEncoding()) {
            case RAW_BYTES:
                return new RawBytesCellDecoder<T>(spec);
            case AVRO:
                return new SpecificAvroCellDecoder<T>(layout, spec);
            case PROTOBUF:
                return new ProtobufCellDecoder<T>(layout, spec);
            default:
                throw new InternalError(
                    "Unhandled cell encoding in reader spec: " + spec.getColumnReaderSpec());
        }
    }

}
